package com.shuhe.domain.dbTodb;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.shuhe.JettyServer;
import com.shuhe.common.MyDateUtils;
import com.shuhe.common.Tools;
import com.shuhe.domain.mxgc.ConnectBean;
import com.shuhe.domain.mxgc.Cqlb;
import com.shuhe.domain.mxgc.TableUpdateRelation;
import com.shuhe.metastore.DBSql;
import com.shuhe.metastore.DbInflux;
import com.shuhe.metastore.MetaMysql;
import org.influxdb.InfluxDB;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.*;

import static com.shuhe.common.CommonFunction.closeDb;
import static com.shuhe.common.CommonFunction.getUUID;
import static com.shuhe.common.MyDateUtils.*;
import static com.shuhe.common.Tools.*;


public class DbToDbThread implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(DbToDbThread.class);
    private Cqlb cqlb;
    private DbConnect dbConnect;


    public DbToDbThread(Cqlb cqlb, DbConnect dbConnect) {
        this.cqlb = cqlb;
        this.dbConnect = dbConnect;
    }

    public void run() {
        String begin_time = MyDateUtils.getNowDate();
        String sql = "";
        try {
            String selectColumn = "";
            //得到源端库的表结构
            selectColumn = getColumnsStr2(cqlb, cqlb.getDb_type(), dbConnect.getSourConn());
            selectColumn = selectColumn.toLowerCase();

            //增量区间转换
            String update_time = cqlb.getStart_time();
            String end_time = cqlb.getNew_time();
            String update_day = cqlb.getTimeadd_interval();
            if (update_day.toLowerCase().indexOf("m") >= 0) {
                update_day = update_day.replace("m", "");
                //分钟
                end_time = getBeforeMinuteBydate(update_time, Integer.parseInt(update_day));
            }
            if (update_day.toLowerCase().indexOf("d") >= 0) {
                update_day = update_day.replace("d", "");
                //天
                end_time = getBeforeDayBydate(update_time, Integer.parseInt(update_day));
            }
            // 判断增量是否要继续执行
            if (cqlb.getCron_type().toLowerCase().equals("timeadd")) {
                String db_end_time = cqlb.getNew_time();
                if (db_end_time == null || db_end_time.equals("")) {
                    db_end_time = getBeforeMinuteM(1);
                    //增量间隔为空，默认为上一分钟
                }
                if (MyDateUtils.timeDifference(end_time, db_end_time) < 0) {  //end_time大于当前时间下一分钟
                    return;
                }
            }
            //获取增量查询带条件sql  获取增量删除带条件sql
            if (cqlb.getDb_type().equals("influxdb")) {
                //influxdb不支持schema_name.table_name的写法
                sql = "select " + selectColumn + " from " + cqlb.getTable_name() + "  where ";
            } else {
                sql = "select " + selectColumn + " from " + cqlb.getFull_table_name() + " a where ";
            }
            //获取增量查询和删除语句的条件
            HashMap<String, String> keyValueArray = getSqlIncrQueryAndDel(cqlb.getTimeadd_column(),
                    cqlb.getTimecolumn_format(), update_time, end_time);
            String whereStr = keyValueArray.get("whereStr");
            String whereStrDel = keyValueArray.get("whereStrDel");
            sql = sql + whereStr;
            String sqlDel = "delete from " + cqlb.getDesttable_name() + " where " + whereStrDel;
            if (cqlb.getCron_type().toLowerCase().equals("all")) {
                if (cqlb.getDb_type().equals("influxdb")) {
                    sql = "select " + selectColumn + " from " + cqlb.getTable_name() + " ";
                } else {
                    sql = "select " + selectColumn + " from " + cqlb.getFull_table_name() + " a";
                }
                sqlDel = "truncate table " + cqlb.getDesttable_name();
            }
            DBSql.execSql(dbConnect.getDestConn(), sqlDel);
            //日志开始
            writeLog(begin_time, "", "", "");

            QueryResult result = null;
            ResultSet rs = null;
            if (cqlb.getDb_type().equals("influxdb")) {
                result = dbConnect.getSourInfluxDB().query(new Query(sql, cqlb.getDb_name()));
            } else {
                rs = DBSql.querySql(dbConnect.getSourConn(), sql);
            }

            //获取目标数据库列们
            ArrayList<String> columns = new ArrayList<>(Arrays.asList(selectColumn.split(",")));
            //目标数据库执行语句集合
            ArrayList sqls = new ArrayList();
            int recordNo = 0;
            //获取批量插入语句的方法
            recordNo = batchExecSqls(cqlb, result, rs, recordNo, sqls, columns);
            LOG.info(cqlb.getDesttable_name() + "记录个数：" + recordNo);
            String log_end_time = MyDateUtils.getNowDate();
            //日志结束
            writeLog(begin_time, log_end_time, "", String.valueOf(recordNo));
            //更新时间标记

            if (cqlb.getCron_type().toLowerCase().equals("timeadd")) {
                sql = "update mxgc_user_task_sjjr set start_time = '" + end_time + "' where id = '" + cqlb.getId() +
                        "'";
                LOG.info(sql);
                MetaMysql.execSql(dbConnect.getMetaConn(), sql);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            //错误日志
            writeLog(begin_time, "", ex.getMessage(), "");
            LOG.error("操作失败:" + sql, ex);
        }
    }

    public void writeLog(String begin_time, String end_time, String error_msg,
                         String record_count) {
        error_msg = error_msg.replaceAll("'", "''");
        if (error_msg.length() > 4000) {
            error_msg = error_msg.substring(0, 4000);
        }
        String sql =
                "select batch_id, task_id from mxgc_user_task_log_sjjr where batch_id = '" + this.cqlb.getBatch_id() + "' " + "and" + "  " + "task_id = '" + this.cqlb.getId() + "'";
        try {
            ResultSet rs = MetaMysql.querySql(dbConnect.getMetaConn(), sql);
            if (rs.next()) {
                String run_status = "";
                if (error_msg.length() > 0) {
                    run_status = "fail";
                } else {
                    run_status = "success";
                }
                sql = "update mxgc_user_task_log_sjjr set  run_endtime = '" + end_time + "', " + "error_msg " + "= '" + error_msg + "', count = '" + record_count + "',run_status='" + run_status + "' where batch_id =" + " '" + this.cqlb.getBatch_id() + "' and task_id = '" + this.cqlb.getId() + "'";
            } else {
                sql = "insert into mxgc_user_task_log_sjjr(batch_id, task_id, run_starttime, run_endtime, error_msg, "
                        + "count," + "task_name,cron_fun) " + "values('" + this.cqlb.getBatch_id() + "', '" + this.cqlb.getId() + "', '" + begin_time + "'," + " '" + end_time + "', '" + error_msg + "', '" + record_count + "','" + this.cqlb.getTask_name() + "','" + this.cqlb.getCronFun() + "')";
            }
            MetaMysql.execSql(dbConnect.getMetaConn(), sql);
        } catch (Exception ex) {
            ex.printStackTrace();
            LOG.error("写入日志失败", ex);
        }
    }


    public HashMap<String, String> getSqlIncrQueryAndDel(String timeadd_column, String timecolumn_format,
                                                         String update_time, String end_time) {
        HashMap<String, String> keyValueArray = new HashMap<>();
        String[] incrementColumns = cqlb.getTimeadd_column().split("!#!");
        String whereStr = "";
        String whereStrDel = "";
        try {
            for (int i = 0; i < incrementColumns.length; i++) {
                if (incrementColumns[i].length() <= 0) {
                    continue;
                }
                if (i == 0) {
                    whereStr =
                            whereStr + " (" + incrementColumns[i] + " >= " + "'" + sjGsh(cqlb.getTimecolumn_format(),
                                    update_time) + "' and " + incrementColumns[i] + " " + "< '" + sjGsh(cqlb.getTimecolumn_format(), end_time) + "') ";
                    whereStrDel =
                            whereStrDel + " (replace(" + incrementColumns[i] + ",'T',' ') >= " + "'" + sjGsh(cqlb.getTimecolumn_format(), update_time) + "' and replace(" + incrementColumns[i] + " ,'T','')" + "< '" + sjGsh(cqlb.getTimecolumn_format(), end_time) + "') ";
                } else {
                    whereStr =
                            whereStr + " or (" + incrementColumns[i] + " >= " + "'" + sjGsh(cqlb.getTimecolumn_format(), update_time) + "' and " + incrementColumns[i] + " " + "< '" + sjGsh(cqlb.getTimecolumn_format(), end_time) + "') ";
                    whereStrDel =
                            whereStrDel + " or (replace(" + incrementColumns[i] + ",'T',' ') >= " + "'" + sjGsh(cqlb.getTimecolumn_format(), update_time) + "' and replace(" + incrementColumns[i] + ",'T',' ') " + "< '" + sjGsh(cqlb.getTimecolumn_format(), end_time) + "') ";
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        keyValueArray.put("whereStr", whereStr);
        keyValueArray.put("whereStrDel", whereStrDel);
        return keyValueArray;
    }


    public int batchExecSqls(Cqlb cqlb, QueryResult result, ResultSet rs, int recordNo,
                             ArrayList sqls, ArrayList<String> columns) throws Exception {
        Integer batchSize = Integer.parseInt(JettyServer.globalProperties.getProperty("batch.size"));
        String sql = "";
        if (cqlb.getDb_type().equals("influxdb")) {
            for (QueryResult.Result r : result.getResults()) {
                if (r.getSeries() == null) {
                    break;
                }
                sql = "insert into " + cqlb.getDesttable_name();
                for (QueryResult.Series series : r.getSeries()) {
                    String fieldStr = "";
                    for (String values : series.getColumns()) {
                        // 列
                        fieldStr = fieldStr + values + ",";
                    }
                    fieldStr = fieldStr.substring(0, fieldStr.length() - 1);
                    for (List<Object> values : series.getValues()) {
                        String valueStr = "";
                        // 数据
                        for (int i = 0; i < values.size(); i++) {
                            String value = "";
                            Object valueO = values.get(i);
                            if (valueO == null) {
                                value = "";
                            } else {
                                value = values.get(i).toString().replace("\"\"", "\"");
                            }
                            if (value == null) {
                                value = "";
                            }
                            valueStr = valueStr + "'" + value + "',";
                        }
                        valueStr = valueStr.substring(0, valueStr.length() - 1);
                        String sqlinsert = sql + "(" + fieldStr + ") values (" + valueStr + ")";
                        sqls.add(sqlinsert);
                        if (sqls.size() >= batchSize) {
                            DBSql.batchExecSql(dbConnect.getDestConn(), sqls);
                            sqls.clear();
                        }
                        recordNo++;
                    }
                }
            }
            if (sqls.size() > 0) {
                DBSql.batchExecSql(dbConnect.getDestConn(), sqls);
                sqls.clear();
            }
        } else {
            while (rs.next()) {
                ResultSetMetaData rsd = rs.getMetaData();
                sql = "insert into " + cqlb.getDesttable_name();
                String fieldStr = "";
                String valueStr = "";
                for (int i = 1; i <= rsd.getColumnCount(); i++) {
                    String columnName = rsd.getColumnName(i).toString().toLowerCase();
                    if (columns.indexOf(columnName) < 0) {
                        continue;
                    }
                    fieldStr = fieldStr + rsd.getColumnName(i) + ",";
                    String value = rs.getString(i);
                    if (value == null) {
                        value = "";
                    }
                    if (rsd.getColumnTypeName(i).indexOf("CHAR") < 0) {
                        if (value.equals("")) {
                            valueStr = valueStr + "null,";
                            continue;
                        }
                    }
                    value = value.replaceAll("'", "''");
                    valueStr = valueStr + "'" + value + "',";
                }
                if (cqlb.getTask_type().equals("stg")) {
                    //接入
                    fieldStr = fieldStr + "inner_id, inner_time";
                    valueStr = valueStr + "'" + getUUID() + "'," + "'" + MyDateUtils.getNowDate() + "'";

                } else {
                    if (cqlb.getTask_type().equals("push")) {
                        //推送
                        fieldStr = fieldStr + "push_id, push_time";
                        valueStr = valueStr + "'" + getUUID() + "'," + "'" + MyDateUtils.getNowDate() + "'";

                    } else {
                        fieldStr = fieldStr.substring(0, fieldStr.length() - 1);
                        valueStr = valueStr.substring(0, valueStr.length() - 1);
                    }
                }


                sql = sql + "(" + fieldStr + ") values (" + valueStr + ")";
                sqls.add(sql);
                if (sqls.size() >= batchSize) {
                    DBSql.batchExecSql(dbConnect.getDestConn(), sqls);
                    sqls.clear();
                }
                recordNo++;
            }
            if (sqls.size() > 0) {
                DBSql.batchExecSql(dbConnect.getDestConn(), sqls);
                sqls.clear();
            }
        }

        return recordNo;
    }
}
